from gai.v2.spark.transformer import DayRectifier
from gai.v2.spark.transformer import FeatureRetriever
from pyspark.context import SparkContext, SparkConf
from pyspark.sql import HiveContext

sparkconf = SparkConf().setAppName("liz_pred").set("spark.ui.showConsoleProgress", "false")
sc = SparkContext(conf=sparkconf)
spark = HiveContext(sc)
sc.setLogLevel('ERROR')
hiveCtx = HiveContext(sc)

db = 'liz'
# bi_ods.xdata_sample_gender
in_tb = '0331_fuy_sens_all_3301'
hiveCtx.sql("use {db}".format(db=db))
df_tmp = hiveCtx.sql("select * from {in_tb}".format(in_tb=in_tb))

df_tmp_pd = df_tmp.toPandas()

df_tmp_pd['day'] = '20200401'
df_tmp_pd['day'] = df_tmp_pd['day'].apply(lambda x: str(x))
df_tmp_pd['matched_gid'] = df_tmp_pd['gid']
df_tmp_df = spark.createDataFrame(df_tmp_pd)

day_rectifier = DayRectifier(inputCol='day', outputCol='rectified_day')
rectified_day_df = day_rectifier.transform(df_tmp_df)
print(rectified_day_df.head())

cols = [
    'ft_2week_rest_active_period_ls',
    'ft_2week_rest_less_act_period_ls',
    'ft_night_stay_detail_ls',
    'ft_month_work_sum_act_times',
    'ft_city_stay_oneday_cnt',
    'ft_2week_work_most_act_times',
    'ft_workplace_stability',
    'ft_month_work_less_act_period_ls',
    'ft_work_cons_ls',
    'ft_month_work_active_period_ls',
    'ft_month_rest_less_act_times',
    'ft_2week_rest_less_act_times',
    'ft_month_work_most_act_times',
    'ft_month_work_less_act_times',
    'ft_dis_label',
    'ft_month_rest_active_period_ls',
    'ft_night_stay_twoday_cnt',
    'ft_2week_work_most_act_period_ls',
    'ft_2week_rest_most_act_times',
    'ft_2week_rest_sum_act_times',
    'ft_2week_work_less_act_period_ls',
    'ft_2week_work_active_period_ls',
    'ft_lbs_ktv_weekly',
    'ft_city_stay_city_ls',
    'ft_month_rest_most_act_period_ls',
    'ft_night_stay_wifimac_cnt',
    'ft_largest_cate_ls',
    'ft_city_stay_twoday_cnt',
    'ft_2week_work_sum_act_times',
    'ft_night_stay_oneday_cnt',
    'ft_pwoi_all_often_consum',
    'ft_pwoi_rest_mostoften',
    'ft_pwoi_all_mostoften',
    'ft_pwoi_hour_mostoften_ls',
    'ft_pwoi_all_often_ls',
    'ft_pwoi_rest_often_ls',
    'ft_pwoi_rest_times_ls',
    'ft_pwoi_hour_often_ls',
    'ft_pwoi_all_times_ls',
    'ft_pwoi_hour_times_ls',
    'ft_pwoi_all_mostoften_consume'
]

feature_retriever = FeatureRetriever(inputIdCol="matched_gid",
                                     inputDayCol="rectified_day",
                                     outputFeatureCols=cols,
                                     extraParams={'span.in.months': 1})

retrieved_feature_df = feature_retriever.transform(rectified_day_df)

print(retrieved_feature_df.head(2))
print(retrieved_feature_df.count())

# retrieved_feature_df.toPandas().to_csv('gender_ios.csv', header=True, index=False, sep='\x01')
retrieved_feature_df.write.saveAsTable('0331_fuy_sens_all_3301_info', None, "overwrite", None)
